package com.crazyice.lee.jobs;

import com.crazyice.lee.writer.JdbcWriter;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;
import org.apache.flink.types.Row;

import java.io.IOException;
import java.io.InputStream;
import java.sql.Types;
import java.util.Properties;

public class Mysql2MysqlJob {

    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try (InputStream inputStream = JdbcWriter.class.getResourceAsStream("/jdbc.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        init();

        //设置环境
        final ExecutionEnvironment environment = properties.getProperty("local.running").equalsIgnoreCase("true") ? ExecutionEnvironment.createLocalEnvironment() : ExecutionEnvironment.getExecutionEnvironment();
        final BatchTableEnvironment tableEnv = BatchTableEnvironment.getTableEnvironment(environment);

        //获取数据-数据源
        TypeInformation[] studentFieldTypes = new TypeInformation[]{
                BasicTypeInfo.INT_TYPE_INFO,
                BasicTypeInfo.STRING_TYPE_INFO,
                BasicTypeInfo.INT_TYPE_INFO,
                BasicTypeInfo.BOOLEAN_TYPE_INFO
        };
        String[] studentFieldNames = new String[]{
                "studentId", "name", "age", "sex"
        };
        DataSource<Row> student = environment.createInput(JDBCInputFormat.buildJDBCInputFormat()
                .setDrivername(properties.getProperty("DriverClass"))
                .setDBUrl(properties.getProperty("SourceDriverUrl"))
                .setUsername(properties.getProperty("SourceUser"))
                .setPassword(properties.getProperty("SourcePassword"))
                .setQuery("select studentId,name,age,sex from student")
                .setRowTypeInfo(new RowTypeInfo(studentFieldTypes, studentFieldNames))
                .finish()
        ).name("读取学生库");

        TypeInformation[] schoolFieldTypes = new TypeInformation[]{
                BasicTypeInfo.INT_TYPE_INFO,
                BasicTypeInfo.INT_TYPE_INFO,
                BasicTypeInfo.INT_TYPE_INFO
        };
        String[] schoolFieldNames = new String[]{
                "schoolId", "orderId", "studentNumber"
        };
        DataSource<Row> school = environment.createInput(JDBCInputFormat.buildJDBCInputFormat()
                .setDrivername(properties.getProperty("DriverClass"))
                .setDBUrl(properties.getProperty("SourceDriverUrl"))
                .setUsername(properties.getProperty("SourceUser"))
                .setPassword(properties.getProperty("SourcePassword"))
                .setQuery("select schoolId,orderId,studentNumber from school")
                .setRowTypeInfo(new RowTypeInfo(schoolFieldTypes, schoolFieldNames))
                .finish()
        ).name("读取学校库");

        TypeInformation[] scoreFieldTypes = new TypeInformation[]{
                BasicTypeInfo.LONG_TYPE_INFO,
                BasicTypeInfo.LONG_TYPE_INFO
        };
        String[] scoreFieldNames = new String[]{
                "studentId", "scoreTotal"
        };
        DataSource<Row> score = environment.createInput(JDBCInputFormat.buildJDBCInputFormat()
                .setDrivername(properties.getProperty("DriverClass"))
                .setDBUrl(properties.getProperty("SourceDriverUrl"))
                .setUsername(properties.getProperty("SourceUser"))
                .setPassword(properties.getProperty("SourcePassword"))
                .setQuery("select studentId,scoreTotal from score")
                .setRowTypeInfo(new RowTypeInfo(scoreFieldTypes, scoreFieldNames))
                .finish()
        ).name("读取考试成绩库");

        TypeInformation[] wishFieldTypes = new TypeInformation[]{
                BasicTypeInfo.LONG_TYPE_INFO,
                BasicTypeInfo.INT_TYPE_INFO,
                BasicTypeInfo.INT_TYPE_INFO
        };
        String[] wishFieldNames = new String[]{
                "studentId", "schoolId", "orderId"
        };
        DataSource<Row> wish = environment.createInput(JDBCInputFormat.buildJDBCInputFormat()
                .setDrivername(properties.getProperty("DriverClass"))
                .setDBUrl(properties.getProperty("SourceDriverUrl"))
                .setUsername(properties.getProperty("SourceUser"))
                .setPassword(properties.getProperty("SourcePassword"))
                .setQuery("select studentId,schoolId,orderId from wish_school")
                .setRowTypeInfo(new RowTypeInfo(wishFieldTypes, wishFieldNames))
                .finish()
        ).name("读取报考志愿库");

        //处理数据
        //1.将DataSet注册为Table
        tableEnv.registerDataSet("student", student);
        tableEnv.registerDataSet("school", school);
        tableEnv.registerDataSet("score", score);
        tableEnv.registerDataSet("wish", wish);

        //2.通过SQL处理数据
        StringBuffer stringBuffer = new StringBuffer();
        stringBuffer.append("SELECT a.schoolId as schoolId,a.studentId as studentId FROM wish as a,score as b,");
        stringBuffer.append(" (select schoolId from school order by orderId limit 10) as c");
        stringBuffer.append(" where a.schoolId=c.schoolId");
        stringBuffer.append(" and b.scoreTotal>500 and a.studentId=b.studentId");
        stringBuffer.append(" order by a.schoolId asc,b.scoreTotal desc,a.orderId asc");

        Table enroll = tableEnv.sqlQuery(stringBuffer.toString()).select("schoolId,studentId");

        //sink
        //1.将Table转换为DataSet
        DataSet<Row> resultSchool = tableEnv.toDataSet(enroll, Row.class);
        //2.输出到数据库
        int[] sqlType = new int[]{
                Types.INTEGER, Types.BIGINT
        };
        resultSchool.output(
                JDBCOutputFormat.buildJDBCOutputFormat()
                        .setDrivername(properties.getProperty("DriverClass"))
                        .setDBUrl(properties.getProperty("SourceDriverUrl"))
                        .setUsername(properties.getProperty("SourceUser"))
                        .setPassword(properties.getProperty("SourcePassword"))
                        .setQuery("insert into enroll (schoolId,studentId) values (?,?)")
                        .setSqlTypes(sqlType)
                        .finish()
        ).name("写入数据库");
        environment.execute("处理招考录取数据");
    }

}
